package com.lagou.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.lagou.dao.OrderDao;
import com.lagou.enumration.OrderStatusEnum;
import com.lagou.model.entity.OrderDO;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderCancelEventConsumer {

    private static final Logger logger = LoggerFactory.getLogger(OrderCancelEventConsumer.class);

    /**
     * 最大重试次数
     */
    private static final int MAX_RETRY_TIMES = 3;

    private Map<Long,AtomicInteger> map = new ConcurrentHashMap<Long, AtomicInteger>();

    @Autowired
    private OrderDao orderDao;

    private ObjectMapper objectMapper = new ObjectMapper();

    @RabbitListener(queues = "queue.order.cancel.dlx")
    public void handle(@Payload String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            logger.info("message:{},deliveryTag:{}", deliveryTag);
            OrderDO orderDO = objectMapper.readValue(message, OrderDO.class);
            Optional<OrderDO> orderDOOptional = orderDao.findById(orderDO.getId());
            if (!orderDOOptional.isPresent()) {
                logger.info("订单不存在，message:{}", message);
                return;
            }

            // 判断是否是未支付状态
            orderDO = orderDOOptional.get();
            if (!OrderStatusEnum.NOT_PAY.getValue().equals(orderDO.getStatus())) {
                logger.info("订单不是未支付状态，message:{}", message);
                return;
            }

            // 更新为未支付状态
            int row = orderDao.updateOrderStatusByStatus(orderDO.getId(),
                    OrderStatusEnum.NOT_PAY.getValue(), OrderStatusEnum.CANCEL.getValue());
            if (row <= 0) {
                logger.info("更新订单为支付状态失败:{}", message);
            }
        } catch (Exception e) {
            // TODO 记录消费次数，超过三次不再重试，持久化消息，然后自动或手动补偿
            logger.error("消息消费失败, message:{}", message);
            AtomicInteger times = map.getOrDefault(deliveryTag, new AtomicInteger(0));
            map.putIfAbsent(deliveryTag, times);
            int temp = times.incrementAndGet();
            if (temp > MAX_RETRY_TIMES) {
                // TODO 记录补偿日志
                try {
                    channel.basicAck(deliveryTag, false);
                } catch (IOException ex) {
                  logger.error("消息确认失败");
                }
            }
            try {
                channel.basicReject(deliveryTag, true);
            } catch (IOException ex) {
                logger.error("消息拒绝失败");
            }
        }

        try {
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
